package cn.wcg.crawler;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class HBaseApi {

    /*
     * create 'data','work','company'
     * describe 'data'
     * alter 'data',{NAME=>'info',VERSION=>3}
     * disable 'data' drop 'data'
     * put 'data','0','work:name', 'zz'
     * scan 'data'
     * get 'stu','0'
     * get 'stu','0','work'
     * get 'stu','0','work:post'
     * scan 'stu',{STARTROW=>'0',STOPROW=>'3'}
     * scan 'stu',{RAW=>true}
     * delete 'stu','1001','info1:sex'
     * delete 'stu','1001','info1'
     * deleteall 'stu','1001'
     * truncate_preserve 'data'
     */

    private static Connection connection = null;
    private static Admin admin = null;

    public static void init() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "master,slave0,slave1");
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = connection.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static boolean is_table_exists(String table_name) throws IOException {
        boolean exists = admin.tableExists(TableName.valueOf(table_name));
        return exists;
    }

    public static void close() {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        if (connection != null) {
            try {
                connection.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    public static void create_table(String table_name, String... cfs) throws IOException {
        if (cfs.length <= 0) {
            System.out.println("列族信息为空");
            return;
        }
        if (is_table_exists(table_name)) {
            System.out.println(table_name + "已存在");
            return;
        }
        HTableDescriptor hTableDescriptor = new HTableDescriptor(TableName.valueOf(table_name));

        for (String cf : cfs) {
            HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(cf);
            hTableDescriptor.addFamily(hColumnDescriptor);
        }

        admin.createTable(hTableDescriptor);
        System.out.println("创建表" + table_name + "成功");
    }

    public static void drop_table(String table_name) throws IOException {
        if (!is_table_exists(table_name)) {
            System.out.println("表不存在");
        }

        admin.disableTable(TableName.valueOf(table_name));

        admin.deleteTable(TableName.valueOf(table_name));
    }

    public static void create_namespace(String ns) {
        NamespaceDescriptor namespaceDescriptor = NamespaceDescriptor.create(ns).build();

        try {
            admin.createNamespace(namespaceDescriptor);
        } catch (NamespaceExistException e) {
            System.out.println(ns + "命名空间已存在");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public static void put_string_data(String table_name, String row_key, String cf, String cn, String value) throws IOException {
        // cf是列族，cn是列名
        Table table = connection.getTable(TableName.valueOf(table_name));

        Put put = new Put(Bytes.toBytes(row_key));

        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn), Bytes.toBytes(value));

        table.put(put);

        table.close();
    }

    public static void put_int_data(String table_name, String row_key, String cf, String cn, Integer value) throws IOException {
        // cf是列族，cn是列名
        Table table = connection.getTable(TableName.valueOf(table_name));

        Put put = new Put(Bytes.toBytes(row_key));

        put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cn), Bytes.toBytes(value));

        table.put(put);

        table.close();
    }

    public static void get_data(String table_name, String row_key, String cf, String cn) throws IOException {
        Table table = connection.getTable(TableName.valueOf(table_name));

        Get get = new Get(Bytes.toBytes(row_key));

        if (!cf.equals("") && cn.equals("")) {
            get.addFamily(Bytes.toBytes(cf));
        } else if (!cn.equals("") && !cf.equals("")) {
            get.addColumn(Bytes.toBytes(cf), Bytes.toBytes("cn"));
        }

        Result result = table.get(get);

        for (Cell cell : result.rawCells()) {
            System.out.println(Bytes.toString(CellUtil.cloneFamily(cell)));
            System.out.println(Bytes.toString(CellUtil.cloneQualifier(cell)));
            System.out.println(Bytes.toString(CellUtil.cloneValue(cell)));
        }

        table.close();
    }

    public static void scan_table(String table_name) throws IOException {
        Table table = connection.getTable(TableName.valueOf(table_name));

        Scan scan = new Scan();
//        Scan scan = new Scan(Bytes.toBytes("1001"), Bytes.toBytes("1003"));

        ResultScanner resultScanner = table.getScanner(scan);

        for (Result result : resultScanner) {
            for (Cell cell : result.rawCells()) {
                System.out.println(Bytes.toString(CellUtil.cloneRow(cell)));
                System.out.println(Bytes.toString(CellUtil.cloneFamily(cell)));
                System.out.println(Bytes.toString(CellUtil.cloneQualifier(cell)));
                String q = Bytes.toString(CellUtil.cloneQualifier(cell));
                if (q.equals("min_wage") || q.equals("max_wage") || q.equals("nums") || q.equals("min_experience") || q.equals("max_experience") || q.equals("age") ){
                    System.out.println(Bytes.toInt(CellUtil.cloneValue(cell)));
                }else {
                    System.out.println(Bytes.toString(CellUtil.cloneValue(cell)));
                }
            }
        }
    }

    public static void get_17(String table_name) throws IOException {
        Table table = connection.getTable(TableName.valueOf(table_name));

        Scan scan = new Scan();
//        Scan scan = new Scan(Bytes.toBytes("1001"), Bytes.toBytes("1003"));

        ResultScanner resultScanner = table.getScanner(scan);

        for (Result result : resultScanner) {
            for (Cell cell : result.rawCells()) {
                String q = Bytes.toString(CellUtil.cloneQualifier(cell));
                if (q.equals("max_experience")){
                    Integer m = Bytes.toInt(CellUtil.cloneValue(cell));
                    if (m == 100){
                        System.out.println(Bytes.toString(CellUtil.cloneRow(cell)));
                    }
                }
            }
        }
    }

    public static List get_column_list(String table_name, String family, String column, String return_type) throws IOException {
        Table table = connection.getTable(TableName.valueOf(table_name));

        Scan scan = new Scan();
//        Scan scan = new Scan(Bytes.toBytes("1001"), Bytes.toBytes("1003"));
        ResultScanner resultScanner = table.getScanner(scan);

        String f;
        String c;
        List results = new ArrayList<>();
        for (Result result : resultScanner) {
            for (Cell cell : result.rawCells()) {
                f = Bytes.toString(CellUtil.cloneFamily(cell));
                c = Bytes.toString(CellUtil.cloneQualifier(cell));
                if (f.equals(family) && c.equals(column)){
                    if (return_type.equals("String")){
                        results.add(Bytes.toString(CellUtil.cloneValue(cell)));
                    }else if (return_type.equals("Integer")){
                        results.add(Bytes.toInt(CellUtil.cloneValue(cell)));
                    }else {
                        System.out.println("没有这样的类型");
                        return null;
                    }
                }
            }
        }
        return results;
    }

    public static void main(String[] args) throws IOException {
        init();
//        get_17("data");
        get_data("data","8","","");
        close();
    }
}
